home *** CD-ROM | disk | FTP | other *** search
/ Skunkware 5 / Skunkware 5.iso / src / Games / ms-0.07 / lib / work.c < prev    next >
Encoding:
C/C++ Source or Header  |  1995-06-27  |  17.3 KB  |  722 lines
  1. /* work.c -- MandelSpawn work distribution */
  2.  
  3. /*
  4.   Queue requests for calculation, possibly from different clients
  5.   (such as multiple windows), distribute them among the computation
  6.   servers and return the results to the client that requested them.
  7.  
  8.   See work.h for some kind of interface definition. 
  9. */
  10.  
  11. /*  
  12.     This file is part of MandelSpawn, a network Mandelbrot program.
  13.  
  14.     Copyright (C) 1990-1993 Andreas Gustafsson
  15.  
  16.     MandelSpawn is free software; you can redistribute it and/or modify
  17.     it under the terms of the GNU General Public License, version 1,
  18.     as published by the Free Software Foundation.
  19.  
  20.     MandelSpawn is distributed in the hope that it will be useful,
  21.     but WITHOUT ANY WARRANTY; without even the implied warranty of
  22.     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  23.     GNU General Public License for more details.
  24.  
  25.     You should have received a copy of the GNU General Public License,
  26.     version 1, along with this program; if not, write to the Free 
  27.     Software Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  28. */
  29.  
  30.  
  31. #include <stdio.h>
  32. #include <string.h>
  33. #include <errno.h>    /* pre-X11R4 systems need this for EWOULDBLOCK */
  34.  
  35. #ifdef NO_STRDUP
  36. char *strdup();
  37. #endif
  38.  
  39. /*
  40.   It is impossible to declare malloc() in a portable way.
  41.   Be prepared to change these declarations. 
  42. */
  43. void *malloc(), *realloc(); 
  44. void free();
  45.  
  46. #include "datarep.h"
  47. #include "ms_ipc.h"
  48. #include "io.h"
  49. #include "work.h"
  50.  
  51. /* This is for Linux */
  52. #ifndef FNDELAY
  53. #ifdef O_NDELAY
  54. #define FNDELAY O_NDELAY
  55. #endif
  56. #endif
  57.  
  58. #ifdef NO_BCOPY
  59. #define bcopy(s,d,n) memcpy(d,s,n)
  60. #endif
  61.  
  62. typedef struct slave
  63. { char *name_string;        /* machine name of the slave */
  64.   NET_ADDRESS name;         /* network address of the slave */
  65.   int has_timeout;         /* the timeout has been initialized */
  66.   unsigned timeout;        /* timeout in milliseconds */
  67. #ifndef OLD_TIMEOUT
  68.   long timeout_at;        /* when to timeout (seconds since epoc) */
  69. #else
  70.   char *timer_id;         /* timeout id (for removing the timeout) */
  71. #endif
  72.   unsigned int n_timeouts;     /* how many times the slave has timed out */
  73.   unsigned int n_packets;    /* number of packets that have arrived */
  74.   unsigned int n_late_packets;    /* number of packets that arrived too late */
  75.   unsigned long mi_count;     /* number of iterations done by this slave */
  76.   unsigned int no;        /* slave serial number */
  77.   struct wf_state *backptr;    /* back pointer to the wf_state */
  78.   int disabled;            /* slave disabled due to error */
  79. } slave;
  80.  
  81. typedef struct chunk
  82. { struct chunk *next;
  83.   struct chunk *prev;
  84.   int drawn;            /* true if at least one reply has arrived */
  85.   char *client;         /* pointer to widget owning this chunk */
  86.   unsigned int no;         /* serial number within current sequence */
  87.   char *client_data;        /* client data (unknown size) */
  88.   char *slave_data;        /* slave data */
  89.   unsigned int slave_datalen;    /* length of slave data */
  90. } chunk;
  91.  
  92. struct wf_state
  93. { 
  94.   io_state *io;            /* pointer to I/O object */
  95.   int sequence;            /* current sequence number */
  96.   int pid;            /* pid of this process */
  97.   unsigned n_slaves;         /* number of slaves */
  98.   struct slave **slaves;    /* array of pointers to slave descriptors */
  99.   int n_chunks;            /* number of chunks in the active sequence */
  100.   unsigned int max_chunks;     /* current size of chunk index */
  101.   struct chunk **chunks;     /* pointer to the chunk index */
  102.   struct chunk to_draw;     /* head of queue of chunks to be drawn */
  103.   struct chunk drawn;        /* head of queue of chunks already drawn */
  104.   struct chunk *insert_point;    /* point in queue for inserting new work */
  105. };
  106.  
  107. /* forward refs */
  108. static void handle_reply_msg(), whip_slave(), timeout_set(), timeout_unset();
  109.  
  110. /* names of files containing server hostnames */
  111. #define PERSONAL_SLAVEFILE ".mslaves" 
  112. #ifndef PUBLIC_SLAVEFILE
  113. #define PUBLIC_SLAVEFILE "/usr/local/etc/mslaves"
  114. #endif
  115.  
  116. #define SERVER_PROG    "mslaved"
  117.  
  118. #define INITIAL_CHUNKS 1024
  119.  
  120. #define MAX_WORKPACKET_SIZE 64
  121. #define DATAGRAM_BYTES 1200
  122.  
  123. char *getenv();
  124.  
  125. #ifdef HAVE_SOCKETS
  126.  
  127. /*
  128.   Get the address of a host given either the host name or the address
  129.   in nnn.nnn.nnn.nnn notation 
  130. */
  131.  
  132. struct hostent *gethostbyname();
  133.  
  134. static struct in_addr *
  135. gethostaddrbywhatever(p)
  136.      char *p;
  137. { struct hostent *hp;
  138.  
  139.   /* statically allocated return value buffer */
  140.   /* (this is ugly, but gethostbyname() is no better) */
  141.   static struct in_addr nobyinaddr;
  142.  
  143.   if ((hp = gethostbyname(p)) != NULL)
  144.   { return((struct in_addr *) (hp->h_addr));
  145.   }
  146.   else
  147.   { /* try it as a number nnn.nnn.nnn.nnn */
  148.     if ((nobyinaddr.s_addr = inet_addr(p)) == -1)
  149.     { return(0);
  150.     }
  151.     return(&nobyinaddr);
  152.   }
  153. }
  154.  
  155.  
  156. /* 
  157.   Null-terminate the current field in a whitespace-separated list
  158.   and return a pointer to the next field or NULL if there is no
  159.   next field.  Text after a hash sign is taken as comment and ignored.
  160. */
  161.  
  162. static char *
  163. next_field(p) char *p;
  164. { while(1)
  165.   { register char c = *p;
  166.     switch(c)
  167.     {
  168.     case ' ':
  169.     case '\t':
  170.       *p = '\0';
  171.       while(1)
  172.       { int c = *++p;
  173.     if(c == '\n' || c == '\0') return(0);
  174.     if(c != ' ' && c != '\t') break;
  175.       }
  176.       return(p);
  177.     case '\0':
  178.       return(0);
  179.     case '\n':
  180.     case '#':
  181.       *p = '\0';
  182.       return(0);
  183.     default:
  184.       break;
  185.     }
  186.     p++;
  187.   }
  188. }
  189. #endif /* HAVE_SOCKETS */
  190.  
  191. slave *new_slave(wf, i)
  192.   wf_state *wf;
  193.   int i;
  194. { slave *s = (slave *) malloc(sizeof(slave));
  195.   s->mi_count = 0L;
  196.   s->has_timeout = 0;
  197.   s->n_timeouts = s->n_packets = s->n_late_packets = 0;
  198.   s->disabled = 0;
  199.   s->no = i;
  200.   s->backptr = wf;
  201.   return(s);
  202. }
  203.  
  204. /* Symbolic names for the ends of a pipe */
  205. #define READ 0
  206. #define WRITE 1
  207.  
  208. /* Initialize the workforce */
  209.  
  210. wf_state *
  211. wf_init(timeout, pipe_mux, socket_mux)
  212.     unsigned timeout;
  213.     io_multiplex pipe_mux, socket_mux;
  214. { FILE *f;
  215.   int i;
  216.   char *filename, *home;
  217.   unsigned int size = 16; /* initial size of slave table */
  218.   char buf[256];
  219.   wf_state *wf = (wf_state *) malloc(sizeof(wf_state));
  220.  
  221.   /* general initialization: */
  222.   
  223.   /* set up the chunk index */
  224.   wf->max_chunks = INITIAL_CHUNKS;
  225.   wf->n_chunks = 0;
  226.   wf->chunks = (chunk **) malloc(wf->max_chunks * sizeof(chunk *));
  227.   
  228.   /* set up the chunk queues */
  229.   wf->to_draw.prev = wf->to_draw.next = &wf->to_draw;
  230.   wf->drawn.prev = wf->drawn.next = &wf->drawn;
  231.  
  232.   wf->pid = getpid();
  233.   wf->sequence = 0;
  234.  
  235. #ifdef HAVE_SOCKETS  
  236.   /* .mslaves file stuff */
  237.   wf->slaves = (slave **) malloc(size*sizeof(slave *));
  238.   home = getenv("HOME");
  239.   if(!home)
  240.     wf_error("HOME not set");
  241.   filename = malloc((unsigned)(strlen(home)+1+strlen(PERSONAL_SLAVEFILE)+1));
  242.   strcpy(filename, home); 
  243.   strcat(filename, "/");
  244.   strcat(filename, PERSONAL_SLAVEFILE);
  245.   f = fopen(filename, "r");
  246.   free(filename);
  247.   if(!f)
  248.   { f = fopen(PUBLIC_SLAVEFILE, "r");
  249.   }
  250.   if(!f) /* no .mslaves file */
  251. #endif /* HAVE_SOCKETS */
  252.   { slave *s;
  253.     int pipe_from_server[2];
  254.     int pipe_to_server[2];
  255. #ifdef HAVE_SOCKETS    
  256.     wf_warn("No .mslaves file found, reverting to single-CPU mode");
  257. #endif /* HAVE_SOCKETS */
  258.     wf->slaves = (slave **) malloc(sizeof(slave *));
  259.     s = new_slave(wf, 0);
  260.     /* s->name is not used */
  261.     s->name_string = "localhost";
  262.     s->timeout = timeout;
  263.     wf->slaves[0] = s;
  264.     wf->n_slaves = 1;
  265.  
  266.     pipe(pipe_to_server);
  267.     pipe(pipe_from_server);
  268.   
  269.     if(! fork())
  270.     { /* child */
  271.       close(0); dup(pipe_to_server[READ]);
  272.       close(1); dup(pipe_from_server[WRITE]);
  273.       close(pipe_to_server[WRITE]);
  274.       close(pipe_from_server[READ]);
  275.       execlp(SERVER_PROG, SERVER_PROG, "-p", "-t0", (char *) 0);
  276.       wf_error("could not exec server program");
  277.     }
  278.  
  279.     wf->io = io_init(
  280.              IO_TRANS_PIPE,
  281.              pipe_mux,
  282.              pipe_from_server[READ], pipe_to_server[WRITE],
  283.              (char *) wf,
  284.              /* this allocates space according to MAX_DATAGRAM_SIZE */
  285.              malloc(sizeof(Message)), sizeof(Message),
  286.              handle_reply_msg, wf_tick);
  287.     close(pipe_to_server[READ]);
  288.     close(pipe_from_server[WRITE]);
  289.   }
  290. #ifdef HAVE_SOCKETS  
  291.   else
  292.   { /* found a .mslaves file */
  293.     i = 0;
  294.     while(1)
  295.     { char *p; /* points to current field in .mslaves line */
  296.       char *q; /* points to next field in .mslaves line */
  297.       slave *s;
  298.       struct in_addr *ina;
  299.       unsigned port;
  300.       
  301.       if(!fgets(buf, sizeof(buf), f))
  302.     break;
  303.       if(buf[0] == '\n' || buf[0] == '#')
  304.     continue;
  305.       p = buf;
  306.       q = next_field(p);
  307.       ina = gethostaddrbywhatever(p);
  308.       
  309.       if(ina == 0)
  310.       {    static char warn[] = "unknown host in .mslaves, ignored: ";
  311.     /* trying to keep up with GNU coding standards... */
  312.     char *msg = malloc(sizeof(warn)+strlen(p));
  313.     strcpy(msg, warn);
  314.     strcat(msg, p);
  315.     wf_warn(msg);
  316.     free(msg);
  317.     continue;
  318.       }
  319.       
  320.       p = q;
  321.       port = DEFAULT_PORT;
  322.       if(p) /* there is a "port" field */
  323.       {    q = next_field(p);
  324.     port = (unsigned) atoi(p);
  325.     if(port == 0) /* probably not an integer, and port 0 is bad anyway */
  326.     { static char warn[] = "bad port field in .mslaves, machine ignored: ";
  327.       char *msg = malloc(sizeof(warn)+strlen(p));
  328.       strcpy(msg, warn);
  329.       strcat(msg, p);
  330.       wf_warn(msg);
  331.       free(msg);
  332.       continue;
  333.     }
  334.       }
  335.       
  336.       if(q)
  337.       {    wf_warn("trailing junk in .mslaves");
  338.       }
  339.       
  340.       if(i>=size) 
  341.       {    size *= 2;
  342.     wf->slaves =
  343.       (slave **) realloc((char *) wf->slaves, size*sizeof(slave *));
  344.       }
  345.       s = new_slave(wf, i);
  346.       bcopy((char *) ina, (char *) &s->name.sin_addr, sizeof(struct in_addr));
  347.       s->name.sin_family = AF_INET;
  348.       s->name.sin_port = htons(port);
  349.       s->name_string = strdup(buf);
  350.       s->timeout = timeout;
  351.       wf->slaves[i] = s;
  352.       i++;
  353.     }
  354.     wf->n_slaves = i;
  355.   }
  356.  
  357.   /* make a socket for communicating with the slaves */
  358.   { int sock;
  359.     if((sock = socket(AF_INET, SOCK_DGRAM, 0)) < 0) 
  360.       wf_error("socket");
  361.     /* make sure we don't block reading the socket */  
  362.     if(fcntl(sock, F_SETFL, FNDELAY) == -1)
  363.       wf_error("unblocking socket");
  364.     wf->io = io_init(IO_TRANS_UDP,
  365.              socket_mux,
  366.              sock, sock, (char *) wf,
  367.              /* this allocates space according to MAX_DATAGRAM_SIZE */
  368.              malloc(sizeof(Message)), sizeof(Message),
  369.              handle_reply_msg, wf_tick);
  370.   }
  371. #endif /* HAVE_SOCKETS */
  372.   return(wf);
  373. }
  374.  
  375.  
  376. /*
  377.   Delete the chunk "c" from anywhere in the queue.  Don't deallocate
  378.   the chunk yet because duplicated messages may still refer to it.  
  379. */
  380.  
  381. static void
  382. queue_delete(c)
  383.      chunk *c;
  384. { c->next->prev = c->prev;
  385.   c->prev->next = c->next;
  386.   c->next = c->prev = NULL; /* just for easier debugging */
  387. }
  388.  
  389.  
  390. /* Add a chunk to the tail of the queue. */
  391.  
  392. static void
  393. queue_add(q, c) chunk *q; chunk *c;
  394. { c->prev = q->prev;
  395.   c->next = q;
  396.   q->prev->next = c;
  397.   q->prev = c;
  398. }
  399.  
  400.  
  401. /* Test the emptiness of a chunk queue. */
  402.  
  403. static int
  404. queue_empty(q)
  405.      chunk *q;
  406. { return(q->next==q && q->prev==q);
  407. }
  408.  
  409.  
  410. /* Return the head of a chunk queue. */
  411.  
  412. static chunk *
  413. queue_head(q)
  414.      chunk *q;
  415. { return(q->next);
  416. }
  417.  
  418.  
  419. void wf_timed_out(client_data)
  420.   char *client_data;
  421. { slave *s = (slave *) client_data;
  422.   s->has_timeout = 0;
  423.   s->n_timeouts++;
  424.   whip_slave(s->backptr, s);
  425. }
  426.  
  427. /* Set a timeout for a slave; this is done when the slave is whipped. */
  428.  
  429. static void
  430. timeout_set(s)
  431.      slave *s;
  432. {
  433. #ifndef OLD_TIMEOUT
  434.   s->timeout_at = time((long *)0) + s->timeout / 1000;
  435. #else
  436.   /* first make sure there isn't a timeout already */
  437.   timeout_unset(s);
  438.   s->timer_id=wf_add_timeout(s->timeout, (char *) s);
  439. #endif
  440.   s->has_timeout=1;
  441. }
  442.  
  443.  
  444. /* Remove a timeout. */
  445.  
  446. static void 
  447. timeout_unset(s)
  448.      slave *s;
  449. {
  450. #ifdef OLD_TIMEOUT  
  451.   if(s->has_timeout)
  452.     wf_remove_timeout(s->timer_id);
  453. #endif
  454.   s->has_timeout=0;
  455. }
  456.  
  457.  
  458. /* Stop the slaves. */
  459.  
  460. static void
  461. stop_slaves(wf)
  462.     wf_state *wf;
  463. { int i;
  464.   for(i=0; i<wf->n_chunks; i++)
  465.   { chunk *c = wf->chunks[i];
  466.     free((char *) (c->client_data));
  467.     free((char *) (c->slave_data));
  468.     free((char *) c);
  469.   }
  470.   /* don't shrink the chunk index array; we probably need it again */
  471.   wf->n_chunks=0;
  472.   wf->sequence++;
  473.   wf->drawn.prev=wf->drawn.next= &wf->drawn;
  474. }
  475.  
  476.  
  477. /* Put the specified slave to work (if there is any) */
  478.  
  479. static void 
  480. whip_slave(wf, s)
  481.   wf_state *wf;
  482.   slave *s;
  483. { struct
  484.   { WhipMessage m;
  485.     char data[MAX_WORKPACKET_SIZE];
  486.   } mm; /* buffer for building the message to send */
  487.   chunk *c;
  488.   
  489.   if(s->disabled || queue_empty(&wf->to_draw))
  490.     return;
  491.  
  492.   c = queue_head(&wf->to_draw);
  493.   mm.m.header.magic = htons(MAGIC);
  494.   mm.m.header.type = htons(WHIP_MESSAGE);
  495.   mm.m.header.version = htons(VERSION);
  496.   mm.m.header.format = htons(DATA_FORMAT);
  497.   mm.m.id.pid = wf->pid;
  498.   mm.m.id.seq = wf->sequence;
  499.   mm.m.id.slave_no = s->no;
  500.   mm.m.id.chunk_no = c->no;
  501.  
  502.   if(c->slave_datalen > MAX_WORKPACKET_SIZE)
  503.     wf_error("work packet too large");
  504.   bcopy(c->slave_data, mm.m.data, c->slave_datalen);
  505.  
  506.   if(io_send(wf->io, (char *)&mm, sizeof(mm.m)+c->slave_datalen,
  507.          (char *) &s->name, (int) sizeof(s->name)) == -1)
  508.   { wf_warn("error sending datagram, use of affected server disabled");
  509.     s->disabled=1; /* consider this slave unusable */
  510.   }
  511.  
  512.   timeout_set(s); 
  513.   /* move the chunk from the head to the tail of the queue */
  514.   queue_delete(c);
  515.   queue_add(&wf->to_draw, c);
  516. }
  517.  
  518.  
  519. /* Handle a reply from a slave. */
  520.  
  521. static void 
  522. handle_reply_msg(closure, msg, msglen)
  523.   char *closure; /* really a wf_state* */
  524.   Message *msg;
  525.   int msglen; /*ARGSUSED*/
  526. { wf_state *wf = (wf_state *) closure;
  527.   char *client;
  528.   slave *s;
  529.   chunk *c;
  530.   
  531.   unsigned int pid = msg->reply.id.pid;
  532.   unsigned int seqno = msg->reply.id.seq;
  533.   unsigned int chunkno = msg->reply.id.chunk_no;
  534.   unsigned int slaveno = msg->reply.id.slave_no;
  535.   int late;
  536.   
  537.   if(pid != wf->pid) return;
  538.   if(slaveno >= wf->n_slaves) return;
  539.   s = wf->slaves[slaveno];
  540.   s->n_packets++;
  541.   if(seqno != wf->sequence) 
  542.   { s->n_late_packets++;
  543.     return;
  544.   }
  545.   if(chunkno >= wf->n_chunks) return;
  546.   c = wf->chunks[chunkno];
  547.   
  548.   timeout_unset(s);
  549.  
  550.   client = c->client;
  551.  
  552.   /*
  553.     The chunk is too late if the client has gone away or if the chunk
  554.     has been drawn already.
  555.   */
  556.   late = (!client || c->drawn);
  557.  
  558.   if(late) /* ignore the message if the chunk arrived too late */
  559.   { s->n_late_packets++;
  560.   }
  561.   else
  562.   { queue_delete(c);
  563.     queue_add(&wf->drawn, c);
  564.     c->drawn=1;
  565.   }
  566.  
  567.   /* Put the slave to work again */
  568.   whip_slave(wf, s);
  569.   
  570.   if(!late)
  571.   { wf_draw(client, c->client_data, (char *) &(msg->reply.data));
  572.     s->mi_count += ntohl(msg->reply.mi_count);
  573.   }
  574.  
  575.   /* If done, stop all the slaves and deallocate the chunks */
  576.   if(queue_empty(&wf->to_draw))
  577.       stop_slaves(wf);
  578. }
  579.  
  580.  
  581. /*
  582.   Prepare for dispatching new chunks.
  583. */
  584.  
  585. void 
  586. wf_begin_dispatch(wf)
  587.      wf_state *wf;
  588. {
  589.   wf->insert_point = wf->to_draw.next;
  590. }
  591.  
  592.  
  593. /*
  594.   This function is called by the Ms widget for each chunk it 
  595.   wants to be calculated.
  596. */
  597.  
  598. void 
  599. wf_dispatch_chunk(wf, client, client_data, client_datalen,
  600.           slave_data, slave_datalen)
  601.      wf_state *wf;
  602.      char *client;
  603.      char *client_data;
  604.      unsigned int client_datalen;
  605.      char *slave_data;
  606.      unsigned int slave_datalen;
  607. { int chunkno = wf->n_chunks++;
  608.   chunk *c = (chunk *) malloc(sizeof(chunk));
  609.   c->client_data = (char *) malloc(client_datalen);
  610.   bcopy(client_data, c->client_data, client_datalen);
  611.   c->slave_data = (char *) malloc(slave_datalen);
  612.   bcopy(slave_data, c->slave_data, slave_datalen);
  613.   c->slave_datalen = slave_datalen;
  614.   c->client = client;
  615.   c->drawn = 0;
  616.   c->no = chunkno;
  617.   /* grow the chunk index if necessary */
  618.   if(chunkno >=  wf->max_chunks)
  619.   { wf->max_chunks *= 2;
  620.     wf->chunks = (chunk **) realloc((char *) wf->chunks,
  621.                        wf->max_chunks*sizeof(chunk *));
  622.   }
  623.   wf->chunks[chunkno] = c;
  624.   queue_add(wf->insert_point, c);
  625. }
  626.  
  627.  
  628. /* Make sure all the slaves are put to work. */
  629.  
  630. void wf_restart(wf)
  631.   wf_state *wf;
  632. { int i;
  633.   for(i=0; i < wf->n_slaves; i++)
  634.   { whip_slave(wf, wf->slaves[i]);
  635.   }
  636. }
  637.  
  638.  
  639. /* Handle the situation of a client aborting prematurely. */
  640.  
  641. void wf_client_died(wf, cli)
  642.   wf_state *wf; char *cli;
  643. { chunk *c;
  644.   chunk *next_c;
  645.   /* Remove all chunks of the dead widget from the work queue */
  646.   /* and nullify their widget pointer fields so that late packets */
  647.   /* won't reference the nonexistent widget */
  648.   for(c=wf->to_draw.next; c != &wf->to_draw; c=next_c)
  649.   { /* need to use a temporary variable because */
  650.     /* relinking the chunk fouls up the .next field */
  651.     next_c=c->next; 
  652.     if(c->client == cli)
  653.     { c->client = NULL;
  654.       queue_delete(c);
  655.       queue_add(&wf->drawn, c);
  656.     }
  657.   }
  658. }
  659.  
  660.  
  661. /* Print performance statistics. */
  662.  
  663. void wf_print_stats(wf, f)
  664.   wf_state *wf; FILE *f;
  665. { int i;
  666.   int active=0;
  667.   unsigned long mi_tot=0;
  668.   fprintf(f, "\n%-22s %10s %10s %10s %10s\n",
  669.      "Host", "iterations", "packets", "timeouts", "late");
  670.   for(i=0; i<wf->n_slaves; i++)
  671.   { slave *s= wf->slaves[i];
  672.     fprintf(f, "%-22s %10lu %10u %10u %10u\n",
  673.        s->name_string, 
  674.        s->mi_count, s->n_packets, s->n_timeouts, s->n_late_packets);
  675.     if(s->mi_count)
  676.       active++;
  677.     mi_tot += s->mi_count;
  678.   }
  679.   fprintf(f, "%d servers, %d active, %lu iterations total\n",
  680.      wf->n_slaves, active, mi_tot);
  681.   fflush(f);
  682. }
  683.  
  684.  
  685. /* Make information about the maximum message size available. */
  686. unsigned
  687. wf_max_message_size()
  688. { return(DATAGRAM_BYTES-sizeof(ReplyHeader));
  689. }
  690.  
  691. #ifndef OLD_TIMEOUT
  692. void wf_tick(closure)
  693.   char *closure;
  694. { wf_state *wf = (wf_state *) closure;
  695.   int i;
  696.   unsigned n_slaves = wf->n_slaves;
  697.   long now = time((long *) 0);
  698.   for(i=0; i<n_slaves; i++)
  699.   { slave *s = s=wf->slaves[i];
  700.     if(s->has_timeout && s->timeout_at <= now)
  701.     { wf_timed_out((char *) s);
  702.     }
  703.   }
  704. }
  705. #endif
  706.  
  707. wf_main(wf)
  708.   wf_state *wf;
  709. { io_main(wf->io);
  710. }
  711.  
  712. void wf_done(wf)
  713.   wf_state *wf;
  714. { io_done(wf->io);
  715. }
  716.  
  717. io_state *wf_get_io(wf)
  718.   wf_state *wf;
  719. {
  720.   return wf->io;
  721. }
  722.